Python Asyncio

参考文档

Asyncio是用来实现异步网络操作，并发，协程

Asyncio关键字

• event_loop 事件循环 程序开启一个事件队列，把一些函数注册到事件队列中，当满足事件发生的条件，调用相应的协程函数
• coroutine 协程 协程对象，指一个使用async关键字定义的函数，当调用函数时，不会立即执行函数，而是返回一个协程对象。协程对象需要注册到事件队列中，由事件队列调用
• task 任务 一个协程对象就是一个原生可以挂起的函数，任务则是对协程进一步封装，其中包含了任务的各种状态
• future 代表将来执行或没有执行的的任务结果，它与task没有本质上的区别
• async/await 关键字，python3.5+用于定义协程的关键字，async定义一个协程，await用于挂起阻塞异步调用接口

定义一个协程

使用async关键字定义协程

from time import time
import asyncio


async def f():
    print("i'm coroutine")

start = time()
coroutine = f()
# 获取一个事件loop
loop = asyncio.get_event_loop()
# 将协程加入到事件队列中
loop.run_until_complete(coroutine)
end = time()
print('Time:{}'.format(end-start))

async关键字定义一个协程，使用get_event_loop()方法获取当前主线程的事件loop；再使用run_until_complete(coroutine)将协程注册到事件队列中，并启用事件轮循。

创建task

协程对象不能直接运行，当注册到事件队列中，其实是run_until_complete(coroutine)将协程封装成一个task任务，用于保存协程的状态，以及未来获取协程结果

from time import time
import asyncio


async def f():
    print("i'm coroutine")
    return "return i'm coroutine"

start = time()
coroutine = f()
# 获取当前主线程的事件轮循队列
loop = asyncio.get_event_loop()
# 创建任务
task = loop.create_task(coroutine)
# 执行任务
loop.run_until_complete(task)
print(task)
print(task.result())
end = time()
print('it takes %d seconds' % (end-start))

创建task后，在task加入事件循环之前为pending状态，当完成后，状态为finished

关于上面通过loop.create_task(coroutine)创建task,同样的可以通过 asyncio.ensure_future(coroutine)创建task

绑定回调

绑定回调，在task执行完成的时候可以获取执行的结果，回调的最后一个参数是future对象，通过该对象可以获取协程返回值

from time import time
import asyncio


async def f():
    print("i'm coroutine")
    return "return i'm coroutine"


def callback(future):
    print('result is {}'.format(future.result()))


start = time()
coroutine = f()
# 获取当前主线程的事件轮循队列
loop = asyncio.get_event_loop()
# 创建任务
task = loop.create_task(coroutine)
# 绑定回调函数
task.add_done_callback(callback)
# 执行任务
loop.run_until_complete(task)
print(task)

end = time()
print('it takes %d seconds' % (end-start))

通过add_done_callback方法给task任务添加回调函数，当task（也可以说是coroutine）执行完成的时候,就会调用回调函数。并通过参数future获取协程执行的结果。这里我们创建 的task和回调里的future对象实际上是同一个对象

阻塞和await

使用async可以定义一个协程，使用await可以针对耗时操作进行挂起，就像生成器的yield，函数让出控制权。协程遇到await，事件轮循会挂起该协程，执行别的协程，直到协程也挂起或者执行完成，再进行下一个协程的执行

耗时的一般是IO操作，例如网络请求，文件读取。模拟IO操作，使用async.sleep()模拟IO操作，协程的目的也是让IO操作异步化

from time import time
import asyncio


async def f():
    print("i'm coroutine")
    await asyncio.sleep(2)
    return "return i'm coroutine"


# def callback(future):
#     print('result is {}'.format(future.result()))


start = time()
coroutine = f()
# 获取当前主线程的事件轮循队列
loop = asyncio.get_event_loop()
# 创建任务
task = loop.create_task(coroutine)
# 执行任务
loop.run_until_complete(task)
print(task.result())

end = time()
print('it takes %d seconds' % (end-start))

并发和并行

并发是指具有多个活动的系统

并行是指用并发使一个系统运行得更快，并行可以在操作系统的多个抽象层次进行运用

所以并行通常是指有多个任务同时进行，并行则是在同一时刻有多个任务同时进行

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)


# def callback(future):
#     print('result is {}'.format(future.result()))


start = time()
coroutine = f(1)
coroutine1 = f(2)
coroutine2 = f(3)
# 获取当前主线程的事件轮循队列
loop = asyncio.get_event_loop()
# 创建任务列表
tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
    coroutine1), asyncio.ensure_future(coroutine2)]
# 执行任务
loop.run_until_complete(asyncio.wait(tasks))
for task in tasks:
    print('result is {}'.format(task.result()))
end = time()
print('it takes %d seconds' % (end-start))

运行结果
i'm coroutine1
i'm coroutine2
i'm coroutine3
result is return i'm coroutine1
result is return i'm coroutine2
result is return i'm coroutine3
it takes 2 seconds

总共时间为2s。2s阻塞时间，所有协程执行完毕，如果是同步执行，至少需要7s。此时运用asyncio模块实现了并发，asyncio.wait(tasks)也可以使用async.gather(*tasks)，前者接受一个列表，后者接收一堆task

协程嵌套

使用async可以定义协程，协程用于耗时的io操作，我们也可以封装更多的io操作过程，这样就实现了嵌套的协程，即一个协程中await了另外一个协程，如此连接起来

单任务

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)


# def callback(future):
#     print('result is {}'.format(future.result()))


async def main():
    coroutine = f(1)
    result = await coroutine
    print(result)
start = time()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
end = time()
print('time is {}'.format(end-start))

运行结果
i'm coroutine1
return i'm coroutine1
time is 2.015313148498535

当在main这个协成里，等待协程f()的return结果，可以使用await f(1)等待执行的结果，并将其赋值给result

多任务

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)



async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    dones, pendings = await asyncio.wait(tasks)
    for result in dones:
        print(result.result())
start = time()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
end = time()
print('time is {}'.format(end-start))

多任务中，要使用asyncio.wait()并发执行任务，当协程完成之后，使用dones，pendings，dones，pendings类型是集合，若是一个参数，则返回的是一个元组，元组中没有result()方法

运行结果
i'm coroutine1
i'm coroutine2
i'm coroutine3
return i'm coroutine1
return i'm coroutine2
return i'm coroutine3
time is 2.012080430984497

使用asyncio.gather(*tasks)会将结果保存至一个List中

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)



async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    dones, pendings = await asyncio.wait(tasks)
    for result in dones:
        print(result.result())
start = time()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
end = time()
print('time is {}'.format(end-start))

运行结果
i'm coroutine1
i'm coroutine2
i'm coroutine3
<class 'list'> ["return i'm coroutine1", "return i'm coroutine2", "return i'm coroutine3"]
return i'm coroutine1
return i'm coroutine2
return i'm coroutine3
time is 2.0033509731292725

不在main协程函数里处理结果，直接返回await的内容，那么最外层的run_until_complete将会返回main协程的结果。 将上述的代码更改为：

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)


async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    return await asyncio.gather(*tasks)
start = time()
loop = asyncio.get_event_loop()
results = loop.run_until_complete(main())
for result in results:
    print(result)
end = time()
print('time is {}'.format(end-start))

运行结果
i'm coroutine1
i'm coroutine2
i'm coroutine3
return i'm coroutine1
return i'm coroutine2
return i'm coroutine3
time is 2.0101969242095947

或者使用asyncio.wait(tasks)将任务挂起，再从已完成的协程中读取result()值

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)


# def callback(future):
#     print('result is {}'.format(future.result()))


async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    return await asyncio.wait(tasks)

start = time()
loop = asyncio.get_event_loop()
dones, pendings = loop.run_until_complete(main())
for result in dones:
    print(result.result())
end = time()
print('time is {}'.format(end-start))

也可以使用asyncio.as_computed

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(2)
    return "return i'm coroutine{}".format(x)


# def callback(future):
#     print('result is {}'.format(future.result()))


async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    for task in asyncio.as_completed(tasks):
        print(await task)

start = time()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
end = time()
print('time is {}'.format(end-start))

协程的停止

future对象有几个状态：

• Pending
• Running
• Done
• Cacelled

from time import time
import asyncio


async def f(x):
    print("i'm coroutine{}".format(x))
    await asyncio.sleep(x)
    return "return i'm coroutine{}".format(x)


# def callback(future):
#     print('result is {}'.format(future.result()))


async def main():
    coroutine = f(1)
    coroutine1 = f(2)
    coroutine2 = f(3)
    tasks = [asyncio.ensure_future(coroutine), asyncio.ensure_future(
        coroutine1), asyncio.ensure_future(coroutine2)]
    return await asyncio.gather(*tasks)
try:
    start = time()
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    for item in result:
        print(item)
    end = time()
    print('time is {}'.format(end-start))
except KeyboardInterrupt:
    print(asyncio.Task.all_tasks())
    for item in asyncio.Task.all_tasks():
        print(item.cancel())
    loop.stop()
    loop.run_forever()
finally:
    loop.close()

启动事件循环之后，马上ctrl+c，会触发run_until_complete的执行异常 KeyBorardInterrupt。然后通过循环asyncio.Task取消future。可以看到输出如下：

运行结果
i'm coroutine1
i'm coroutine2
i'm coroutine3
{<Task pending coro=<main() running at app.py:29> wait_for=<_GatheringFuture pending cb=[<TaskWakeupMethWrapper object at 0x0000014E4532CD98>()]>>, <Task finished coro=<f() done, defined at app.py:13> result="return i'm coroutine1">, <Task pending coro=<f() running at app.py:15> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x0000014E4533F4F8>()]> cb=[gather.<locals>._done_callback() at C:\Python3\lib\asyncio\tasks.py:691]>, <Task pending coro=<f() running at app.py:15> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x0000014E4556E708>()]> cb=[gather.<locals>._done_callback() at C:\Python3\lib\asyncio\tasks.py:691]>}
True
False
True
True

True表示cannel成功，loop stop之后还需要再次开启事件循环，最后在close，不然还会抛出异常

循环task，逐个cancel是一种方案，可是正如上面我们把task的列表封装在main函数中，main函数外进行事件循环的调用。这个时候，main相当于最外出的一个task，那么处理包装的main函数即可。

不同线程的事件轮循

很多时候，我们的事件循环用于注册协程，而有的协程需要动态的添加到事件循环中。一个简单的方式就是使用多线程。当前线程创建一个事件循环，然后在新建一个线程，在新线程中启动事件循环。当前线程不会被block。

import asyncio
from threading import Thread
import time


def now(): return time.time()


def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()


def more_work(x):
    print('More work {}'.format(x))
    time.sleep(x)
    print('Finished more work {}'.format(x))


start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))

new_loop.call_soon_threadsafe(more_work, 6)
new_loop.call_soon_threadsafe(more_work, 3)

启动上述代码之后，当前线程不会被block，新线程中会按照顺序执行call_soon_threadsafe方法注册的more_work方法， 后者因为time.sleep操作是同步阻塞的，因此运行完毕more_work需要大致6 + 3

新线程协程

import asyncio
import time
from threading import Thread

now = lambda :time.time()


def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

async def do_some_work(x):
    print('Waiting {}'.format(x))
    await asyncio.sleep(x)
    print('Done after {}s'.format(x))


start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))

asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)

上述的例子，主线程中创建一个new_loop，然后在另外的子线程中开启一个无限事件循环。 主线程通过run_coroutine_threadsafe新注册协程对象。这样就能在子线程中进行事件循环的并发操作，同时主线程又不会被block。一共执行的时间大概在6s左右